package com.game.framework.mqsystem.event;

import com.alibaba.cloud.nacos.NacosDiscoveryProperties;
import com.game.framework.common.utils.TopicUtil;
import com.game.framework.common.utils.UUIDUtil;
import com.game.framework.mqsystem.mq.GameMQTemplate;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.core.env.Environment;

import java.lang.reflect.Method;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
public class GameEventContext {

	private NacosDiscoveryProperties nacosDiscoveryProperties;

	@Autowired
	private GameMQTemplate gameMQTemplate;
	private String consumeGroupName;
	private ApplicationContext applicationContext;
	private Map<Integer, Class<? extends IGameEvent>> eventMessageMap = new HashMap<>();
	private Map<Integer, List<GameEventListenerMapping>> gameEventDispatcherMap = new HashMap<>();
	private int localServerId;//本地serverId，用于记录事件是从哪个服务发出的
	private static final String GatewayTopicPrefix = "gateway";

	public GameEventContext(int localServerId, ApplicationContext applicationContext) {
		init(localServerId,null, applicationContext);
	}

	/**
	 * 这个方法可以手动设置消费者组，如果groupName为null，将使用spring.application.name做为消费者组参数
	 */
	public void init(int localServerId,String groupName,ApplicationContext context) {
		this.localServerId = localServerId;
		this.consumeGroupName = groupName;
		this.applicationContext = context;
		//this.gameMQTemplate = applicationContext.getBean(GameMQTemplate.class);
		this.nacosDiscoveryProperties = applicationContext.getBean(NacosDiscoveryProperties.class);
		scanGameEvent();
		consumeEventMessage();

	}

	private void scanGameEvent() {
		String[] beanNames = this.applicationContext.getBeanNamesForAnnotation(GameEventHandler.class);
		for (String beanName : beanNames) {
			Object bean = this.applicationContext.getBean(beanName);
			Method[] methods = bean.getClass().getMethods();
			for (Method method : methods) {
				addEventDispatcherInfo(bean, method);
			}
		}
	}

	/**
	 * 获取所有的本服务要处理的eventId
	 */
	private Collection<Integer> getAllEventIds() {
		return this.eventMessageMap.keySet();
	}

	private void addEventDispatcherInfo(Object target, Method method) {
		GameEventMapping gameEventMapping = method.getAnnotation(GameEventMapping.class);
		if (gameEventMapping != null) {
			GameEventMetaData gameEventMetadata = gameEventMapping.value().getAnnotation(GameEventMetaData.class);
			if (gameEventMetadata != null) {
				int eventId = gameEventMetadata.eventId();
				List<GameEventListenerMapping> gameEventListenerMappings = this.gameEventDispatcherMap.get(eventId);
				if (gameEventListenerMappings == null) {
					gameEventListenerMappings = new ArrayList<>();
					this.gameEventDispatcherMap.put(eventId, gameEventListenerMappings);
				}
				if (this.eventMessageMap.containsKey(eventId)) {
					Class<? extends IGameEvent> oldClazz = this.eventMessageMap.get(eventId);
					if (oldClazz.getName().equals(gameEventMapping.value().getName())) {
						throw new IllegalArgumentException("事件对象 " + oldClazz.getName() + ","
								+ gameEventMapping.value().getName() + " eventId重复！！");
					}
				} else {
					this.eventMessageMap.put(eventId, gameEventMapping.value());
				}
				GameEventListenerMapping gameEventListenerMapping = new GameEventListenerMapping(target, method);
				gameEventListenerMappings.add(gameEventListenerMapping);
			} else {
				throw new IllegalArgumentException(
						"MQ 事件对象没有添加元数据信息,请在事件对象上添加@GameEventMetadata信息，事件对象：" + gameEventMapping.value().getName());
			}
		}
	}

	private IGameEvent getEventMessageInstance(int eventId) {
		Class<? extends IGameEvent> clazz = this.eventMessageMap.get(eventId);
		if (clazz != null) {
			try {
				Object instance = clazz.newInstance();
				return (IGameEvent) instance;
			} catch (InstantiationException | IllegalAccessException e) {
				log.error("创建EventMessage实例失败,eventId:{}", eventId, e);
			}
		}
		return null;
	}

	public void sendEvent(IGameEvent eventMessage) {
		GameEventHeader header = eventMessage.getHeader();
		String id = UUIDUtil.getUID();
		header.setId(id);
		header.setFromServerId(this.localServerId);
		byte[] data = eventMessage.write();
		String tag = String.valueOf(eventMessage.getHeader().getEventId());
		List<String> tags = Arrays.asList(tag);
		String topic = TopicUtil.generateTopic(nacosDiscoveryProperties.getNamespace(), GatewayTopicPrefix);
		this.gameMQTemplate.asyncSendOrderly(topic, data, tags);
		log.debug("发送MQ事件，topic:{}, event:{}, tags:{}", topic, eventMessage, tag);
	}

	private void dispatcherEvent(IGameEvent gaMessage) {
		GameEventHeader header = gaMessage.getHeader();
		List<GameEventListenerMapping> gameEventListenerMappings = this.gameEventDispatcherMap.get(header.getEventId());
		if (gameEventListenerMappings != null) {
			gameEventListenerMappings.forEach(c -> {
				try {
					c.getMethod().invoke(c.getTarget(), gaMessage);
				} catch (Throwable e) {
					log.error("Game event call {}#{}失败", c.getTarget().getClass().getName(), c.getMethod().getName(),
							e);
				}
			});
		}
	}

	private void consumeEventMessage() {
		String topicContent = TopicUtil.generateTopic(nacosDiscoveryProperties.getNamespace(), GatewayTopicPrefix, this.localServerId) ;
		Environment environment = this.applicationContext.getEnvironment();
		RocketMQProperties rocketMQProperties = this.applicationContext.getBean(RocketMQProperties.class);
		// 设置消息者组，同一组下面监听的topic只会被消费一次，防止多次消费
		if (this.consumeGroupName == null || this.consumeGroupName.length() == 0) {
			this.consumeGroupName = environment.getProperty("spring.application.name") + "-EventConsumerGroup";
		}
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(this.consumeGroupName);
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
		consumer.setMessageModel(MessageModel.CLUSTERING);
		consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
		Collection<Integer> eventIds = getAllEventIds();
		if (eventIds.size() > 0) {
			Collection<String> tags = eventIds.stream().map(c -> String.valueOf(c)).collect(Collectors.toList());
			// 选择消息，只接收这些tags的消息
			String subExpression = GameMQTemplate.convertTags(tags);
			try {
				consumer.subscribe(topicContent, subExpression);
				//添加消费监听
				consumer.registerMessageListener(new MessageListenerOrderly() {

					@Override
					public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
						byte[] bytes=null;
						for(MessageExt message : msgs) {
							bytes = message.getBody();
							if(bytes != null) {
								ByteBuf byteBuf = Unpooled.wrappedBuffer(bytes);
								int eventId = byteBuf.readInt();
								IGameEvent eventMessage = getEventMessageInstance(eventId);
								try {
									eventMessage.read(byteBuf);
									//分发消息，注意在这里是在消费者线程中执行的。
									dispatcherEvent(eventMessage);
								} catch (Exception e) {
									log.error("MQ事件处理异常，eventId:{}",eventId,e);
								}
							}
						}
						return ConsumeOrderlyStatus.SUCCESS;
					}
				});
				consumer.start();
			} catch (MQClientException e) {
				log.error("启动事件RocketMQ消费者失败", e);
			}
			log.info("MQ 事件系统启动成功,GroupName:{},topic:{},tags：{}", this.consumeGroupName, topicContent, subExpression);
		}
	}
	

}
